package car_test;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

public class mr_fenxi {
    public static class GenderMapper extends Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text gender = new Text();

        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            try {
                // 按GBK编码读取行数据（根据实际数据编码调整）
                String line = new String(value.getBytes(), 0, value.getLength(), "GBK");
                // 分割CSV数据（支持带引号的字段）
                String[] fields = line.trim().split(",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)", -1);

                // 检查数据完整性（假设性别列是第3列，根据实际数据调整索引）
                if (fields != null && fields.length == 39) {
                    String genderStr = fields[38].trim();  // 关键修改：获取性别列数据

                    // 数据清洗：过滤空白/无效性别
                    if (genderStr.isEmpty() || genderStr.equalsIgnoreCase("null") ||
                            !genderStr.matches("^(男性|女性)$")) {  // 仅保留"男性"/"女性"
                        return;
                    }

                    gender.set(genderStr);
                    context.write(gender, one);  // 输出(性别, 1)
                }
            } catch (Exception e) {
                context.getCounter("DataError", "InvalidLine").increment(1);  // 统计错误行
            }
        }
    }

    public static class GenderReducer extends Reducer<Text, IntWritable, Text, DoubleWritable> {
        private Map<Text, Long> genderCount = new HashMap<>();
        private long total = 0;

        @Override
        public void reduce(Text key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            long sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            total += sum;
            genderCount.put(new Text(key), sum);  // 存储性别计数
        }

        @Override
        public void cleanup(Context context) throws IOException, InterruptedException {
            Set<Text> genders = genderCount.keySet();
            for (Text g : genders) {
                double ratio = (double) genderCount.get(g) / total;  // 计算比例
                context.write(new Text(g.toString() + "人数比例"), new DoubleWritable(ratio));
                context.write(new Text(g.toString() + "人数"), new DoubleWritable(genderCount.get(g)));
            }
        }
    }

    public static void main(String[] args) throws Exception {
        // 添加运行开始提示
        System.out.println("==================== 任务开始运行 ====================");
        System.out.println("开始时间：" + new java.util.Date().toString());

        String namenodeIp = "192.168.128.130";
        String hdfs = "hdfs://" + namenodeIp + ":8020";
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", hdfs);

        Job job = Job.getInstance(conf, "gender_analysis_v2");
        job.setJarByClass(mr_fenxi.class);

        // Mapper配置
        job.setMapperClass(GenderMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // Reducer配置
        job.setReducerClass(GenderReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);

        // 输入输出路径
        String inputPath = hdfs + "/car/data.csv";
        String outputPath = hdfs + "/Output/v2";

        FileInputFormat.addInputPath(job, new Path(inputPath));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        // 清理旧输出目录
        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(new Path(outputPath))) {
            fs.delete(new Path(outputPath), true);
        }

        // 执行作业
        boolean success = job.waitForCompletion(true);

        // 添加运行结束提示
        System.out.println("==================== 任务运行结束 ====================");
        System.out.println("结束时间：" + new java.util.Date().toString());
        System.out.println("任务状态：" + (success ? "成功" : "失败"));

        System.exit(success ? 0 : 1);
    }

}
